package com.zxz.mq.config;

import com.alibaba.fastjson.JSON;
import com.zxz.mq.prop.BindingProperties;
import com.zxz.mq.prop.ExchangeProperties;
import com.zxz.mq.prop.MyMqProperties;
import com.zxz.mq.prop.QueueProperties;
import com.zxz.mq.service.IMessageSendService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@ConditionalOnProperty(name = "my.mq.enable",havingValue = "true")
@Configuration
@Slf4j
public class MyMqConfig {
    @Autowired
    private MyMqProperties myMqProperties;

    @Autowired
    private IMessageSendService messageSendService;

    @Bean("myConnectionFactory")
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(myMqProperties.getHost());
        connectionFactory.setPort(myMqProperties.getPort());
        connectionFactory.setPassword(myMqProperties.getPassword());
        connectionFactory.setUsername(myMqProperties.getUsername());
        connectionFactory.setVirtualHost(myMqProperties.getVhost());
        connectionFactory.setSimplePublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

    @Bean("myRabbitTemplate")
    public RabbitTemplate myRabbitTemplate(@Qualifier("myConnectionFactory") ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
                    log.info("correlationData:{}, ack:{}, cause:{}", correlationData, ack, cause);
                    if (ack && null != correlationData) {
                        String messageId = correlationData.getId();
                        log.info("消息已正确投递到交换机, messageId:{}", messageId);
                        messageSendService.messageSendSuccess(messageId);
                    } else {
                        log.error("消息投递至交换机失败,correlationData:{}，cause:{}", correlationData, cause);
                    }
                }
        );

        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.error("消息无法路由,message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}", message, replyCode, replyText, exchange, routingKey);
            //消息发送失败 触发Return机制
            messageSendService.messageSendReturn(message.getMessageProperties().getMessageId(),exchange,routingKey,new String(message.getBody()));
        });
        return rabbitTemplate;
    }

    @Bean("myRabbitAdmin")
    public RabbitAdmin myRabbitAdmin(@Qualifier("myRabbitTemplate") RabbitTemplate rabbitTemplate) {
        return initRabbitAdmin(rabbitTemplate, myMqProperties.getExchanges(), myMqProperties.getQueues(), myMqProperties.getBindings());
    }

    private RabbitAdmin initRabbitAdmin(RabbitTemplate rabbitTemplate, List<ExchangeProperties> exchanges, List<QueueProperties> queues, List<BindingProperties> bindings) {
        RabbitAdmin admin = new RabbitAdmin(rabbitTemplate);
        Map<String, Exchange> exchangeMap = new HashMap<>(16);
        Map<String, Queue> queueMap = new HashMap<>(16);
        this.setExchange(exchanges,exchangeMap,admin);

        this.setQueue(queues,queueMap, admin);

        this.setBinding(bindings,exchangeMap,queueMap,admin);

        return admin;
    }

    private void setExchange(List<ExchangeProperties> exchanges, Map<String, Exchange> exchangeMap, RabbitAdmin admin) {
        if(CollectionUtils.isEmpty(exchanges)){
            return;
        }

        for (ExchangeProperties exchangeProperties : exchanges){
            if(StringUtils.isEmpty(exchangeProperties.getName())){
                continue;
            }

            String type = exchangeProperties.getType();
            if (StringUtils.isEmpty(type)) {
                log.error("exchangeProperties[{}] type is empty", exchangeProperties.getName());
                continue;
            }
            Map<String, Object> argsMap = new HashMap<>(16);
            if (!StringUtils.isEmpty(exchangeProperties.getArguments())) {
                argsMap.putAll(JSON.parseObject(exchangeProperties.getArguments(), HashMap.class));
            }

            Exchange exchange;

            switch (type){
                case ExchangeTypes.DIRECT:
                    exchange = new DirectExchange(exchangeProperties.getName(), exchangeProperties.isDurable(),exchangeProperties.isAutoDelete(), argsMap);
                    break;
                case ExchangeTypes.FANOUT:
                    exchange = new FanoutExchange(exchangeProperties.getName(), exchangeProperties.isDurable(),exchangeProperties.isAutoDelete(), argsMap);
                    break;
                case ExchangeTypes.TOPIC:
                    exchange = new TopicExchange(exchangeProperties.getName(), exchangeProperties.isDurable(),exchangeProperties.isAutoDelete(), argsMap);
                    break;
                default:
                    log.error("exchangeProperties[{}] type is error", exchangeProperties.getName());
                    continue;
            }

            admin.declareExchange(exchange);
            exchangeMap.put(exchangeProperties.getName(),exchange);
        }
    }

    private void setQueue(List<QueueProperties> queues, Map<String, Queue> queueMap, RabbitAdmin admin) {
        if (CollectionUtils.isEmpty(queues)) {
            return;
        }
        for (QueueProperties queueProperties : queues) {
            if (StringUtils.isEmpty(queueProperties.getName())) {
                continue;
            }
            Map<String, Object> argsMap = new HashMap<>();
            if (!StringUtils.isEmpty(queueProperties.getArguments())) {
                argsMap.putAll(JSON.parseObject(queueProperties.getArguments(), HashMap.class));
            }
            Queue queue = new Queue(queueProperties.getName(), queueProperties.isDurable(), queueProperties.isExclusive(),queueProperties.isAutoDelete(), argsMap);
            admin.declareQueue(queue);
            queueMap.put(queueProperties.getName(),queue);
        }
    }

    private void setBinding(List<BindingProperties> bindings, Map<String, Exchange> exchangeMap, Map<String, Queue> queueMap, RabbitAdmin admin) {
        if(CollectionUtils.isEmpty(bindings)){
            return;
        }

        for (BindingProperties bindingProperties : bindings) {
            String routeKey = bindingProperties.getRouteKey();
            String queueName = bindingProperties.getQueue();
            String exchangeName = bindingProperties.getExchange();

            if (StringUtils.isEmpty(queueName) || StringUtils.isEmpty(exchangeName)) {
                log.error("bind for queue[{}] or exchange[{}] is empty", queueName, exchangeName);
                continue;
            }

            Exchange exchange = exchangeMap.get(exchangeName);
            if (null == exchange) {
                log.error("Exchange[{}] is null.", routeKey);
                continue;
            }

            Queue queue = queueMap.get(queueName);
            if (queue == null) {
                log.error("Queue[{}] is null.", queueName);
                continue;
            }
            if (exchange instanceof TopicExchange) {
                TopicExchange topExchange = (TopicExchange) exchange;
                admin.declareBinding(BindingBuilder.bind(queue).to(topExchange).with(routeKey));
            } else if (exchange instanceof FanoutExchange) {
                FanoutExchange fanoutExchange = (FanoutExchange) exchange;
                admin.declareBinding(BindingBuilder.bind(queue).to(fanoutExchange));
            } else if (exchange instanceof DirectExchange) {
                DirectExchange directExchange = (DirectExchange) exchange;
                admin.declareBinding(BindingBuilder.bind(queue).to(directExchange).with(routeKey));
            }else{
                log.error("setBinding[{}] error.", queueName);
            }
        }
    }
}
